package net.lmc.realtime.jtp.dwd.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class JtpOrderSkuAggrehateDwsJob {
    public static void main(String[] args) {
        //环境
        TableEnvironment tabEnv = getTableEnv();

        //输入表
        createInputTable(tabEnv);

        //处理数据
        Table reportTable = handle(tabEnv);

        createOutputTable(tabEnv);

        saveToDoris(tabEnv,reportTable);
    }

    private static void saveToDoris(TableEnvironment tabEnv, Table reportTable) {
        tabEnv.createTemporaryView("report_table",reportTable);

        tabEnv.executeSql("INSERT INTO dws_trade_sku_order_report_doris_sink\n" +
                "SELECT\n" +
                "    SUBSTRING(CAST(start_time AS STRING), 0, 19) AS start_time,\n" +
                "    SUBSTRING(CAST(end_time AS STRING), 0, 19) AS end_time,\n" +
                "    SUBSTRING(CAST(start_time AS STRING), 0, 10) AS cur_date,\n" +
                "    sku_id, sku_name, spu_id, spu_name, tm_id, tm_name,\n" +
                "    category3_id, category3_name, category2_id, category2_name, category1_id, category1_name,\n" +
                "    order_count, order_amount\n" +
                "FROM report_table");
    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE dws_trade_sku_order_report_doris_sink(\n" +
                "    `start_time` STRING,\n" +
                "    `end_time` STRING,\n" +
                "    `cur_date` STRING,\n" +
                "    `sku_id` STRING,\n" +
                "    `sku_name` STRING,\n" +
                "    `spu_id` STRING,\n" +
                "    `spu_name` STRING,\n" +
                "    `tm_id` STRING,\n" +
                "    `tm_name` STRING,\n" +
                "    `category3_id` STRING,\n" +
                "    `category3_name` STRING,\n" +
                "    `category2_id` STRING,\n" +
                "    `category2_name` STRING,\n" +
                "    `category1_id` STRING,\n" +
                "    `category1_name` STRING,\n" +
                "    `order_count` BIGINT,\n" +
                "    `order_amount` DECIMAL(16, 2)\n" +
                ") WITH (    \n" +
                "'connector' = 'doris',    \n" +
                "'fenodes' = 'node102:8030',    \n" +
                "'table.identifier' = 'jtp_mall_report.dws_trade_sku_order_report',    \n" +
                "'username' = 'root',    \n" +
                "'password' = '123456',    \n" +
                "'sink.batch.interval' = '10s',    \n" +
                "'sink.max-retries' = '3',    \n" +
                "'sink.batch.size' = '1000'\n" +
                ")");
    }

    private static Table handle(TableEnvironment tabEnv) {
        Table order_join_table = tabEnv.sqlQuery("SELECT\n" +
                "    t1.id,\n" +
                "    t1.order_id,\n" +
                "    t1.user_id,\n" +
                "    t1.sku_id,\n" +
                "    t1.sku_name,\n" +
                "    t1.province_id,\n" +
                "    t1.create_time,\n" +
                "    t1.source_type_name,\n" +
                "    t1.sku_num,\n" +
                "    t1.split_total_amount,\n" +
                "    t2.spu_id,\n" +
                "    t2.category3_id,\n" +
                "    t2.tm_id\n" +
                "FROM dwd_order_detail_kafka_source t1\n" +
                "    LEFT JOIN dim_sku_info_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t2        \n" +
                "ON t1.sku_id = t2.row_key");

        tabEnv.createTemporaryView("order_join_table",order_join_table);

        Table result_table = tabEnv.sqlQuery("SELECT\n" +
                "    DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') AS start_time,\n" +
                "    DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') AS end_time,\n" +
                "    sku_id, sku_name, spu_id, category3_id, tm_id ,    \n" +
                "count(distinct id) AS order_count,\n" +
                "    sum(cast(split_total_amount AS DECIMAL(10, 2))) AS order_amount,\n" +
                "    PROCTIME() AS proc_time\n" +
                "FROM TABLE (\n" +
                "    TUMBLE(\n" +
                "         DATA => TABLE order_join_table,\n" +
                "         TIMECOL => DESCRIPTOR(create_time),\n" +
                "         SIZE => INTERVAL '1' MINUTES    \n" +
                ")\n" +
                ")\n" +
                "WHERE user_id IS NOT NULL AND source_type_name IS NOT NULL\n" +
                "GROUP BY\n" +
                "    window_start, window_end,\n" +
                "    sku_id, sku_name, spu_id, category3_id, tm_id");

        tabEnv.createTemporaryView("result_table",result_table);

        Table report_table = tabEnv.sqlQuery("SELECT\n" +
                "    t1.start_time,\n" +
                "    t1.end_time,\n" +
                "    t1.sku_id,\n" +
                "    t1.sku_name,\n" +
                "    t1.spu_id,\n" +
                "    t2.spu_name,\n" +
                "    t1.tm_id,\n" +
                "    t3.tm_name,\n" +
                "    t1.category3_id,\n" +
                "    t4.name AS category3_name,\n" +
                "    t4.category2_id,\n" +
                "    t5.name AS category2_name,\n" +
                "    t5.category1_id,\n" +
                "    t6.name AS category1_name,\n" +
                "    t1.order_count,\n" +
                "    t1.order_amount,\n" +
                "    UNIX_TIMESTAMP() * 1000 AS ts\n" +
                "FROM result_table t1\n" +
                "    LEFT JOIN dim_spu_info_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t2 ON t1.spu_id = t2.row_key\n" +
                "    LEFT JOIN dim_base_trademark_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t3 ON t1.tm_id = t3.row_key\n" +
                "    LEFT JOIN dim_base_category3_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t4 ON t1.category3_id = t4.row_key\n" +
                "    LEFT JOIN dim_base_category2_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t5 ON t4.category2_id = t5.row_key\n" +
                "    LEFT JOIN dim_base_category1_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t6 ON t5.category1_id = t6.row_key");


        return report_table;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE dwd_order_detail_kafka_source\n" +
                "(\n" +
                "    `id`                    STRING,\n" +
                "    `order_id`              STRING,\n" +
                "    `user_id`               STRING,\n" +
                "    `order_status`          STRING,\n" +
                "    `sku_id`                STRING,\n" +
                "    `sku_name`              STRING,\n" +
                "    `province_id`           STRING,\n" +
                "    `activity_id`           STRING,\n" +
                "    `activity_rule_id`      STRING,\n" +
                "    `coupon_id`             STRING,\n" +
                "    `date_id`               STRING,\n" +
                "    `create_time`           TIMESTAMP(3),\n" +
                "    `operate_date_id`       STRING,\n" +
                "    `operate_time`          STRING,\n" +
                "    `source_id`             STRING,\n" +
                "    `source_type`           STRING,\n" +
                "    `source_type_name`      STRING,\n" +
                "    `sku_num`               STRING,\n" +
                "    `split_original_amount` STRING,\n" +
                "    `split_activity_amount` STRING,\n" +
                "    `split_coupon_amount`   STRING,\n" +
                "    `split_total_amount`    STRING,\n" +
                "    `row_op_ts`             STRING,\n" +
                "    `proc_time` AS PROCTIME(),\n" +
                "    WATERMARK FOR create_time AS create_time - INTERVAL '0' MINUTE\n" +
                ") WITH (    \n" +
                "'connector' = 'kafka',    \n" +
                "'topic' = 'dwd-order-detail',    \n" +
                "'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',    \n" +
                "'properties.group.id' = 'gid-dws-trade-sku-order',    \n" +
                "'scan.startup.mode' = 'earliest-offset',    \n" +
                "'format' = 'json',    \n" +
                "'json.fail-on-missing-field' = 'false',    \n" +
                "'json.ignore-parse-errors' = 'true'\n" +
                ")");


        tabEnv.executeSql("CREATE TABLE dim_sku_info_hbase_source (\n" +
                "    row_key STRING,\n" +
                "    info ROW<id STRING, spu_id STRING, price STRING, sku_name STRING, sku_desc STRING, weight STRING, tm_id STRING, category3_id STRING, sku_default_img STRING, is_sale STRING, create_time STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (    \n" +
                "'connector' = 'hbase-2.2',    \n" +
                "'table-name' = 'dim_sku_info',    \n" +
                "'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',    \n" +
                "'lookup.async' = 'true',    \n" +
                "'lookup.cache.max-rows' = '10',    \n" +
                "'lookup.cache.ttl' = '1 hour'\n" +
                ")");

        tabEnv.executeSql("CREATE TABLE dim_spu_info_hbase_source (\n" +
                "    row_key STRING,\n" +
                "    info ROW<id STRING, spu_name STRING, description STRING, category3_id STRING, tm_id STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (    \n" +
                "'connector' = 'hbase-2.2',    \n" +
                "'table-name' = 'dim_spu_info',    \n" +
                "'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',    \n" +
                "'lookup.async' = 'true',    \n" +
                "'lookup.cache.max-rows' = '10',    \n" +
                "'lookup.cache.ttl' = '1 hour'\n" +
                ")");

        tabEnv.executeSql("CREATE TABLE dim_base_trademark_hbase_source (\n" +
                "    row_key STRING,\n" +
                "    info ROW<id STRING, tm_name STRING, logo_url STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (    \n" +
                "'connector' = 'hbase-2.2',    \n" +
                "'table-name' = 'dim_base_trademark',    \n" +
                "'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',    \n" +
                "'lookup.async' = 'true',    \n" +
                "'lookup.cache.max-rows' = '10',    \n" +
                "'lookup.cache.ttl' = '1 hour'\n" +
                ")");

        tabEnv.executeSql("CREATE TABLE dim_base_category3_hbase_source (\n" +
                "    row_key STRING,\n" +
                "    info ROW<id STRING, name STRING, category2_id STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (    \n" +
                "'connector' = 'hbase-2.2',    \n" +
                "'table-name' = 'dim_base_category3',    \n" +
                "'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',    \n" +
                "'lookup.async' = 'true',    \n" +
                "'lookup.cache.max-rows' = '10',    \n" +
                "'lookup.cache.ttl' = '1 hour'\n" +
                ")");

        tabEnv.executeSql("CREATE TABLE dim_base_category2_hbase_source (\n" +
                "    row_key STRING,\n" +
                "    info ROW<id STRING, name STRING, category1_id STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (    \n" +
                "'connector' = 'hbase-2.2',    \n" +
                "'table-name' = 'dim_base_category2',    \n" +
                "'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',    \n" +
                "'lookup.async' = 'true',    \n" +
                "'lookup.cache.max-rows' = '10',    \n" +
                "'lookup.cache.ttl' = '1 hour'\n" +
                ")");

        tabEnv.executeSql("CREATE TABLE dim_base_category1_hbase_source (\n" +
                "    row_key STRING,\n" +
                "    info ROW<id STRING, name STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (    \n" +
                "'connector' = 'hbase-2.2',    \n" +
                "'table-name' = 'dim_base_category1',    \n" +
                "'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',    \n" +
                "'lookup.async' = 'true',    \n" +
                "'lookup.cache.max-rows' = '10',    \n" +
                "'lookup.cache.ttl' = '1 hour'\n" +
                ")");

    }

    private static TableEnvironment getTableEnv() {
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings) ;
        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 3返回对象
        return tabEnv;
    }
}
